#include <sstream>
#include <stdexcept>
#include <string>
#include <utility>

#include "common/config.h"
#include "common/exception.h"
#include "common/logger.h"
#include "common/macros.h"
#include "common/rid.h"
#include "storage/index/b_plus_tree.h"
#include "storage/index/index_iterator.h"
#include "storage/page/b_plus_tree_header_page.h"
#include "storage/page/b_plus_tree_page.h"
#include "storage/page/page_guard.h"
#include "type/value.h"

namespace bustub {

INDEX_TEMPLATE_ARGUMENTS
BPLUSTREE_TYPE::BPlusTree(std::string name, page_id_t header_page_id, BufferPoolManager *buffer_pool_manager,
                          const KeyComparator &comparator, int leaf_max_size, int internal_max_size)
    : index_name_(std::move(name)),
      bpm_(buffer_pool_manager),
      comparator_(std::move(comparator)),
      leaf_max_size_(leaf_max_size),
      internal_max_size_(internal_max_size),
      header_page_id_(header_page_id) {
  WritePageGuard guard = bpm_->FetchPageWrite(header_page_id_);
  auto root_page = guard.AsMut<BPlusTreeHeaderPage>();
  root_page->root_page_id_ = INVALID_PAGE_ID;
}

/*
 * Helper function to decide whether current b+tree is empty
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::IsEmpty() const -> bool {
  auto header_page = bpm_->FetchPageRead(header_page_id_);
  // auto ttt = header_page.As<BPlusTreeHeaderPage>()->root_page_id_;
  // std::cout << ttt << std::endl;
  return header_page.As<BPlusTreeHeaderPage>()->root_page_id_ == INVALID_PAGE_ID;
}
/*****************************************************************************
 * SEARCH
 *****************************************************************************/
/*
 * Return the only value that associated with input key
 * This method is used for point query
 * @return : true means key exists
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::GetValue(const KeyType &key, std::vector<ValueType> *result, Transaction *txn) -> bool {
  if (IsEmpty()) {
    return false;
  }
  // Declaration of context instance.
  auto header_guarded = bpm_->FetchPageRead(header_page_id_);
  if (header_guarded.As<BPlusTreeHeaderPage>()->root_page_id_ == INVALID_PAGE_ID) {
    return false;
  }

  auto root_page = bpm_->FetchPageRead(header_guarded.As<BPlusTreeHeaderPage>()->root_page_id_);
  header_guarded.Drop();

  while (!(root_page.template As<BPlusTreePage>()->IsLeafPage())) {
    int index = root_page.template As<InternalPage>()->Search(key, comparator_);
    if (index == -1) {
      index = root_page.template As<BPlusTreePage>()->GetSize();
    }
    BUSTUB_ASSERT(index > 0, "index <= 0 !");

    auto next_page_id = root_page.template As<InternalPage>()->ValueAt(index - 1);
    root_page = bpm_->FetchPageRead(next_page_id);
  }

  int index = root_page.template As<LeafPage>()->Search(key, comparator_);
  if (index == -1) {
    index = root_page.template As<LeafPage>()->GetSize();
  }
  index--;
  if (comparator_(root_page.template As<LeafPage>()->KeyAt(index), key) == 0) {
    result->push_back(root_page.template As<LeafPage>()->ValueAt(index));
    return true;
  }

  return false;
}

/*****************************************************************************
 * INSERTION
 *****************************************************************************/
/*
 * Insert constant key & value pair into b+ tree
 * if current tree is empty, start new tree, update root page id and insert
 * entry, otherwise insert into leaf page.
 * @return: since we only support unique key, if user try to insert duplicate
 * keys return false, otherwise return true.
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Insert(const KeyType &key, const ValueType &value, Transaction *txn) -> bool {
  // Declaration of context instance.
  Context ctx;
  auto header_guarded = bpm_->FetchPageWrite(header_page_id_);
  // if tree is empty
  if (header_guarded.As<BPlusTreeHeaderPage>()->root_page_id_ == INVALID_PAGE_ID) {
    page_id_t newpage_id = INVALID_PAGE_ID;
    bpm_->NewPageGuarded(&newpage_id);
    BUSTUB_ASSERT(newpage_id != INVALID_PAGE_ID, "New page failed");
    auto newpage_guard = bpm_->FetchPageWrite(newpage_id);
    auto newpage = newpage_guard.AsMut<LeafPage>();
    newpage->Init(leaf_max_size_);
    newpage->Insert(0, key, value);
    auto header_page = header_guarded.AsMut<BPlusTreeHeaderPage>();
    header_page->root_page_id_ = newpage_id;
    return true;
  }

  auto root_page = bpm_->FetchPageWrite(header_guarded.AsMut<BPlusTreeHeaderPage>()->root_page_id_);
  // 向下查找对应的叶子节点
  while (!(root_page.template As<BPlusTreePage>()->IsLeafPage())) {
    if (root_page.template As<BPlusTreePage>()->GetSize() < internal_max_size_) {
      // internal page是安全的
      header_guarded.Drop();
      ctx.write_set_.clear();
    }
    int index = root_page.template As<InternalPage>()->Search(key, comparator_);
    if (index == -1) {
      index = root_page.template As<BPlusTreePage>()->GetSize();
    }
    BUSTUB_ASSERT(index > 0, "index <= 0 !");
    auto next_page_id = root_page.template As<InternalPage>()->ValueAt(index - 1);
    ctx.write_set_.emplace_back(std::move(root_page));
    root_page = bpm_->FetchPageWrite(next_page_id);
  }

  BUSTUB_ASSERT(root_page.template As<BPlusTreePage>()->IsLeafPage(), "root page here should be leaf");
  auto leaf_page_ptr = root_page.template AsMut<LeafPage>();

  // 寻找插入位置
  int insert_pos = leaf_page_ptr->Search(key, comparator_);
  if (insert_pos == -1) {
    insert_pos = leaf_page_ptr->GetSize();
  }

  if (comparator_(key, leaf_page_ptr->KeyAt(insert_pos - 1)) == 0) {
    // LOG_INFO("key already exists!");
    return false;
  }

  if (leaf_page_ptr->GetSize() == leaf_max_size_ - 1) {
    // need split
    page_id_t newpage_id = INVALID_PAGE_ID;
    bpm_->NewPageGuarded(&newpage_id);
    BUSTUB_ASSERT(newpage_id != INVALID_PAGE_ID, "new page failed");
    auto newpage_guard = bpm_->FetchPageWrite(newpage_id);
    auto new_page_ptr = newpage_guard.template AsMut<LeafPage>();

    new_page_ptr->Init(leaf_max_size_);
    new_page_ptr->SetNextPageId(leaf_page_ptr->GetNextPageId());
    leaf_page_ptr->SetNextPageId(newpage_id);

    leaf_page_ptr->Insert(insert_pos, key, value);
    int leaf_page_size = leaf_page_ptr->GetSize();
    auto mid_key = leaf_page_ptr->KeyAt(leaf_page_size / 2);
    for (int i = leaf_page_size / 2; i < leaf_page_size; ++i) {
      new_page_ptr->PushBack(leaf_page_ptr->KeyAt(i), leaf_page_ptr->ValueAt(i));
    }
    leaf_page_ptr->SetSize(leaf_page_size / 2);
    // leaf_page_ptr->IncreaseSize(0-(leaf_page_size-(leaf_page_size/2)));
    // 向父节点插入索引key，并递归的判断父internal节点是否需要分裂
    return InsertInternalRecur(ctx, mid_key, newpage_id, header_guarded);
  }
  // simple insert

  BUSTUB_ASSERT(insert_pos < leaf_max_size_, "insert位置计算错误");
  leaf_page_ptr->Insert(insert_pos, key, value);
  return true;
}

/*****************************************************************************
 * REMOVE
 *****************************************************************************/
/*
 * Delete key & value pair associated with input key
 * If current tree is empty, return immediately.
 * If not, User needs to first find the right leaf page as deletion target, then
 * delete entry from leaf page. Remember to deal with redistribute or merge if
 * necessary.
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::Remove(const KeyType &key, Transaction *txn) {
  // Declaration of context instance.
  Context ctx;
  // get header_page
  auto header_guarded = bpm_->FetchPageWrite(header_page_id_);
  auto header_guarded_ptr = &header_guarded;
  // If empty then return
  if (header_guarded.As<BPlusTreeHeaderPage>()->root_page_id_ == INVALID_PAGE_ID) {
    return;
  }

  // 首先查找对应的key
  auto root_page = bpm_->FetchPageWrite(header_guarded.As<BPlusTreeHeaderPage>()->root_page_id_);
  page_id_t next_page_id = INVALID_PAGE_ID;
  while (!(root_page.template As<BPlusTreePage>()->IsLeafPage())) {
    if (root_page.template As<BPlusTreePage>()->GetSize() > root_page.template As<BPlusTreePage>()->GetMinSize()) {
      header_guarded.Drop();
      header_guarded_ptr = nullptr;
      ctx.write_set_.clear();
    }
    int index = root_page.template As<InternalPage>()->Search(key, comparator_);
    if (index == -1) {
      index = root_page.template As<BPlusTreePage>()->GetSize();
    }
    BUSTUB_ASSERT(index > 0, "index <= 0 !");

    next_page_id = root_page.template As<InternalPage>()->ValueAt(index - 1);
    ctx.write_set_.emplace_back(std::move(root_page));
    root_page = bpm_->FetchPageWrite(next_page_id);
  }

  int index = root_page.template As<LeafPage>()->Search(key, comparator_);
  if (index == -1) {
    index = root_page.template As<LeafPage>()->GetSize();
  }
  index--;
  if (comparator_(root_page.template As<LeafPage>()->KeyAt(index), key) == 0) {
    // remove array_[index]
    ctx.write_set_.emplace_back(std::move(root_page));
    RemoveEntry(ctx, key, header_guarded_ptr);
  }
  // else : 删除的key不存在
}

/*****************************************************************************
 * INDEX ITERATOR
 *****************************************************************************/
/*
 * Input parameter is void, find the leftmost leaf page first, then construct
 * index iterator
 * @return : index iterator
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Begin() -> INDEXITERATOR_TYPE {
  if (this->IsEmpty()) {
    return this->End();
  }
  auto t_page_guarded = bpm_->FetchPageRead(GetRootPageId());
  t_page_guarded = bpm_->FetchPageRead(GetRootPageId());
  while (!(t_page_guarded.template As<BPlusTreePage>()->IsLeafPage())) {
    auto next_page_id = t_page_guarded.template As<InternalPage>()->ValueAt(0);
    t_page_guarded = bpm_->FetchPageRead(next_page_id);
  }

  return {std::move(t_page_guarded), 0, bpm_};
}

/*
 * Input parameter is low key, find the leaf page that contains the input key
 * first, then construct index iterator
 * @return : index iterator
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Begin(const KeyType &key) -> INDEXITERATOR_TYPE {
  if (this->IsEmpty()) {
    return this->End();
  }
  auto t_page_guarded = bpm_->FetchPageRead(GetRootPageId());
  while (!(t_page_guarded.template As<BPlusTreePage>()->IsLeafPage())) {
    auto page_ptr = t_page_guarded.template As<InternalPage>();
    auto index = page_ptr->Search(key, comparator_);
    if (index == -1) {
      index = page_ptr->GetSize();
    }
    index--;
    auto next_page_id = page_ptr->ValueAt(index);
    t_page_guarded = bpm_->FetchPageRead(next_page_id);
  }
  auto page_ptr = t_page_guarded.template As<LeafPage>();
  auto index = page_ptr->Search(key, comparator_);
  if (index == -1) {
    index = page_ptr->GetSize();
  }
  index--;
  if (index < 0) {
    return this->End();
  }
  if (comparator_(page_ptr->KeyAt(index), key) != 0) {
    return this->End();
  }
  return {std::move(t_page_guarded), index, bpm_};
}

/*
 * Input parameter is void, construct an index iterator representing the end
 * of the key/value pair in the leaf node
 * @return : index iterator
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::End() -> INDEXITERATOR_TYPE { return {ReadPageGuard(), -1, nullptr}; }

/**
 * @return Page id of the root of this tree
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::GetRootPageId() -> page_id_t {
  auto header_page = bpm_->FetchPageRead(header_page_id_);
  return header_page.As<BPlusTreeHeaderPage>()->root_page_id_;
}

/*****************************************************************************
 * SELF MADE HELPER FUNC
 *****************************************************************************/

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::InsertInternalRecur(Context &ctx, KeyType key, page_id_t value, WritePageGuard &header_guarded)
    -> bool {
  // 如果没有父节点了，那么创建一个internal节点，存入索引，设置左右孩子，然后设置为根节点
  if (ctx.write_set_.empty()) {
    // add implementation
    int new_page_id = INVALID_PAGE_ID;
    bpm_->NewPageGuarded(&new_page_id);
    BUSTUB_ASSERT(new_page_id != INVALID_PAGE_ID, "new page failed");
    auto new_page_guarded = bpm_->FetchPageWrite(new_page_id);
    auto new_page_ptr = new_page_guarded.AsMut<InternalPage>();
    new_page_ptr->Init(internal_max_size_);
    new_page_ptr->PushBack(key, header_guarded.As<BPlusTreeHeaderPage>()->root_page_id_);
    new_page_ptr->PushBack(key, value);
    BUSTUB_ASSERT(new_page_ptr->GetSize() == 2, "new page has wrong size");
    // SetRootPageId(new_page_id);
    header_guarded.AsMut<BPlusTreeHeaderPage>()->root_page_id_ = new_page_id;
    return true;
  }
  BUSTUB_ASSERT(!ctx.write_set_.empty(), "write set should not be empty");
  auto internal_page_guarded = std::move(ctx.write_set_.back());
  ctx.write_set_.pop_back();
  auto internal_page_ptr = internal_page_guarded.AsMut<InternalPage>();
  if (internal_page_ptr->GetSize() == internal_max_size_) {
    // 内部节点需要分裂
    // 先创建新的internalpage
    page_id_t new_page_id = INVALID_PAGE_ID;
    bpm_->NewPageGuarded(&new_page_id);
    BUSTUB_ASSERT(new_page_id != INVALID_PAGE_ID, "new page failed");
    auto new_page_guarded = bpm_->FetchPageWrite(new_page_id);
    auto new_page_ptr = new_page_guarded.template AsMut<InternalPage>();
    new_page_ptr->Init(internal_max_size_);

    // 寻找在internalpage中的插入位置
    int pos = internal_page_ptr->Search(key, comparator_);
    if (pos == -1) {
      pos = internal_page_ptr->GetSize();
    }
    BUSTUB_ASSERT(pos > 0, "pos<=0");
    if (pos > 1) {
      BUSTUB_ASSERT(comparator_(internal_page_ptr->KeyAt(pos - 1), key) != 0, "existing key should not be here");
    }
    // size-1是internalpage中key的数量
    int k = (internal_page_ptr->GetSize() - 1) / 2;

    // 判断中位数
    if (pos <= k) {
      // array[k]为中位数
      new_page_ptr->PushBack(key, internal_page_ptr->ValueAt(k));
      for (int i = k + 1; i <= internal_page_ptr->GetSize() - 1; ++i) {
        new_page_ptr->PushBack(internal_page_ptr->KeyAt(i), internal_page_ptr->ValueAt(i));
      }
      auto mid_key = internal_page_ptr->KeyAt(k);
      // 丢弃k-1个元素之后的部分
      internal_page_ptr->Truncate(k);
      internal_page_ptr->Insert(pos, key, value);
      // 向父节点继续加入索引
      return InsertInternalRecur(ctx, mid_key, new_page_id, header_guarded);
    }

    if (pos > k + 1) {
      // array[k+1]为中位数
      new_page_ptr->PushBack(key, internal_page_ptr->ValueAt(k + 1));
      for (int i = k + 2; i <= internal_page_ptr->GetSize() - 1; ++i) {
        new_page_ptr->PushBack(internal_page_ptr->KeyAt(i), internal_page_ptr->ValueAt(i));
      }
      new_page_ptr->Insert(pos - k - 1, key, value);
      // 丢弃k个元素之后的部分
      internal_page_ptr->Truncate(k + 1);
      return InsertInternalRecur(ctx, internal_page_ptr->KeyAt(k + 1), new_page_id, header_guarded);
    }

    if (pos == k + 1) {
      // 要插入的key为索引
      new_page_ptr->PushBack(key, value);
      for (int i = k + 1; i <= internal_page_ptr->GetSize() - 1; ++i) {
        new_page_ptr->PushBack(internal_page_ptr->KeyAt(i), internal_page_ptr->ValueAt(i));
      }
      internal_page_ptr->Truncate(k + 1);
      return InsertInternalRecur(ctx, key, new_page_id, header_guarded);
    }

    LOG_INFO("pos is ilegal");
    return false;
  }
  // 不需要分裂，直接加入索引
  int pos = internal_page_ptr->Search(key, comparator_);
  if (pos == -1) {
    pos = internal_page_ptr->GetSize();
  }
  internal_page_ptr->Insert(pos, key, value);
  return true;
}

INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::RemoveEntry(Context &ctx, KeyType key, WritePageGuard *header_guarded) {
  BUSTUB_ASSERT(!ctx.write_set_.empty(), "ctx shouldn't be empty");
  auto &node_guarded = ctx.write_set_.back();
  // delete key from node
  if (node_guarded.As<BPlusTreePage>()->IsLeafPage()) {
    node_guarded.AsMut<LeafPage>()->Delete(key, comparator_);
  } else {
    node_guarded.AsMut<InternalPage>()->Delete(key, comparator_);
  }
  // if node is the root and has only one remaining child
  if (header_guarded != nullptr && node_guarded.PageId() == header_guarded->As<BPlusTreeHeaderPage>()->root_page_id_) {
    if ((!node_guarded.As<BPlusTreePage>()->IsLeafPage()) && node_guarded.As<BPlusTreePage>()->GetSize() == 1) {
      auto node_ptr = node_guarded.As<InternalPage>();
      auto old_root_id = node_guarded.PageId();
      auto child_page_id = node_ptr->ValueAt(0);
      BUSTUB_ASSERT(child_page_id != INVALID_PAGE_ID, "应该作为唯一的有效节点");
      // make node's child the new root
      header_guarded->AsMut<BPlusTreeHeaderPage>()->root_page_id_ = child_page_id;
      // unpin the old root page
      ctx.write_set_.pop_back();
      // delete the old root page
      bpm_->DeletePage(old_root_id);
      // BUSTUB_ASSERT(is_deleted, "bpm delete page fail");
    } else if (node_guarded.As<BPlusTreePage>()->IsLeafPage() && node_guarded.As<BPlusTreePage>()->GetSize() == 0) {
      auto only_leaf_id = node_guarded.PageId();
      ctx.write_set_.pop_back();
      bpm_->DeletePage(only_leaf_id);
      header_guarded->AsMut<BPlusTreeHeaderPage>()->root_page_id_ = INVALID_PAGE_ID;
    }
    return;
  }
  int node_size = node_guarded.As<BPlusTreePage>()->GetSize();
  int node_min_size = node_guarded.As<BPlusTreePage>()->GetMinSize();
  int node_max_size = node_guarded.As<BPlusTreePage>()->GetMaxSize();
  // 叶子节点应该用max_size-1来判断是否合并
  if (node_guarded.As<BPlusTreePage>()->IsLeafPage()) {
    node_max_size--;
  }
  BUSTUB_ASSERT(node_max_size > 0, "wrong max size");
  if (node_size < node_min_size) {
    BUSTUB_ASSERT(ctx.write_set_.size() >= 2, "");
    auto &parent_node_guarded = *(ctx.write_set_.end() - 2);
    // get index in parent
    auto parent_ptr = parent_node_guarded.AsMut<InternalPage>();
    auto node_index = parent_ptr->ValueIndex(node_guarded.PageId());

    // check right sibling
    if (node_index < parent_ptr->GetSize() - 1) {
      auto right_sibling_id = parent_ptr->ValueAt(node_index + 1);
      auto parent_key = parent_ptr->KeyAt(node_index + 1);
      auto right_sibling_guarded = bpm_->FetchPageWrite(right_sibling_id);
      auto right_sibling_size = right_sibling_guarded.template As<BPlusTreePage>()->GetSize();
      if (node_size + right_sibling_size <= node_max_size) {
        // 可以和右侧兄弟合并的情况
        if (node_guarded.As<BPlusTreePage>()->IsLeafPage()) {
          // 对于叶子节点，将右侧节点的值append到左侧节点
          auto node_ptr = node_guarded.AsMut<LeafPage>();
          auto right_sibling_ptr = right_sibling_guarded.template As<LeafPage>();
          for (int i = 0; i < right_sibling_size; ++i) {
            node_ptr->PushBack(right_sibling_ptr->KeyAt(i), right_sibling_ptr->ValueAt(i));
          }
          node_ptr->SetNextPageId(right_sibling_ptr->GetNextPageId());
        } else {
          // 对于internal节点，先把parent中的key append到左侧，再append右侧节点中的值
          auto node_ptr = node_guarded.AsMut<InternalPage>();
          auto right_sibling_ptr = right_sibling_guarded.template As<InternalPage>();
          node_ptr->PushBack(parent_key, right_sibling_ptr->ValueAt(0));
          for (int i = 1; i < right_sibling_size; ++i) {
            node_ptr->PushBack(right_sibling_ptr->KeyAt(i), right_sibling_ptr->ValueAt(i));
          }
        }
        ctx.write_set_.pop_back();
        // node_guarded,node_ptr已失效
        right_sibling_guarded.Drop();
        auto is_deleted = bpm_->DeletePage(right_sibling_id);
        BUSTUB_ASSERT(is_deleted, "delete failed");
        RemoveEntry(ctx, parent_key, header_guarded);
        // 把右侧节点删除
        // BUSTUB_ASSERT(right_sibling_guarded.template As<BPlusTreePage>()->GetSize()==0, "wrong size");

        return;
      }
      // 需要进行redistribution
      if (node_guarded.As<BPlusTreePage>()->IsLeafPage()) {
        auto node_ptr = node_guarded.AsMut<LeafPage>();
        auto right_sibling_ptr = right_sibling_guarded.template AsMut<LeafPage>();
        BUSTUB_ASSERT(right_sibling_ptr->GetSize() > 1, "wrong size");
        node_ptr->PushBack(right_sibling_ptr->KeyAt(0), right_sibling_ptr->ValueAt(0));
        right_sibling_ptr->Delete(0);
        parent_ptr->SetKeyAt(node_index + 1, right_sibling_ptr->KeyAt(0));

      } else {
        // remove right_sibling arr[0].value arr[1].key arr[0].value = arr[1].value t=arr[1].key
        auto node_ptr = node_guarded.AsMut<InternalPage>();
        auto right_sibling_ptr = right_sibling_guarded.template AsMut<InternalPage>();
        BUSTUB_ASSERT(right_sibling_ptr->GetSize() > 1, "wrong size");
        auto moving_up_key = right_sibling_ptr->KeyAt(1);
        auto moving_left_value = right_sibling_ptr->ValueAt(0);
        right_sibling_ptr->Delete(0);

        node_ptr->PushBack(parent_key, moving_left_value);

        parent_ptr->SetKeyAt(node_index + 1, moving_up_key);
      }
      return;
    }

    // check left sibling
    if (node_index > 0) {
      auto left_sibling_id = parent_ptr->ValueAt(node_index - 1);
      auto parent_key = parent_ptr->KeyAt(node_index);
      auto left_sibling_guarded = bpm_->FetchPageWrite(left_sibling_id);
      auto left_sibling_size = left_sibling_guarded.template As<BPlusTreePage>()->GetSize();
      // 可以与左侧兄弟合并的情况
      if (node_size + left_sibling_size <= node_max_size) {
        if (node_guarded.As<BPlusTreePage>()->IsLeafPage()) {
          // 对于叶子节点，将自己的值append到左侧兄弟节点
          auto node_ptr = node_guarded.AsMut<LeafPage>();
          auto left_sibling_ptr = left_sibling_guarded.template AsMut<LeafPage>();
          for (int i = 0; i < node_size; ++i) {
            left_sibling_ptr->PushBack(node_ptr->KeyAt(i), node_ptr->ValueAt(i));
          }
          left_sibling_ptr->SetNextPageId(node_ptr->GetNextPageId());
        } else {
          // 对internal节点，先把parent中的key append到左侧，再append右侧节点中的值
          auto node_ptr = node_guarded.AsMut<InternalPage>();
          auto left_sibling_ptr = left_sibling_guarded.template AsMut<InternalPage>();
          left_sibling_ptr->PushBack(parent_key, node_ptr->ValueAt(0));
          BUSTUB_ASSERT(node_size == node_ptr->GetSize(), "size changed");
          for (int i = 1; i < node_size; ++i) {
            left_sibling_ptr->PushBack(node_ptr->KeyAt(i), node_ptr->ValueAt(i));
          }
        }
        // BUSTUB_ASSERT(node_guarded.As<BPlusTreePage>()->GetSize()==0, "wrong size");
        auto deleting_id = node_guarded.PageId();
        node_guarded.Drop();
        auto is_deleted = bpm_->DeletePage(deleting_id);
        BUSTUB_ASSERT(is_deleted, "delete failed");
        ctx.write_set_.pop_back();
        RemoveEntry(ctx, parent_key, header_guarded);
        return;
      }

      // 需要进行redistribution
      if (node_guarded.As<BPlusTreePage>()->IsLeafPage()) {
        auto node_ptr = node_guarded.AsMut<LeafPage>();
        auto left_sibling_ptr = left_sibling_guarded.template AsMut<LeafPage>();
        BUSTUB_ASSERT(left_sibling_size > 1, "wrong size");
        BUSTUB_ASSERT(left_sibling_size == left_sibling_ptr->GetSize(), "wrong size");
        node_ptr->Insert(0, left_sibling_ptr->KeyAt(left_sibling_size - 1),
                         left_sibling_ptr->ValueAt(left_sibling_size - 1));
        left_sibling_ptr->Delete(left_sibling_size - 1);
        // left_sibling_size失效
        parent_ptr->SetKeyAt(node_index, node_ptr->KeyAt(0));
      } else {
        auto node_ptr = node_guarded.AsMut<InternalPage>();
        auto left_sibling_ptr = left_sibling_guarded.template AsMut<InternalPage>();
        BUSTUB_ASSERT(left_sibling_size == left_sibling_ptr->GetSize(), "wrong size");

        auto moving_up_key = left_sibling_ptr->KeyAt(left_sibling_size - 1);
        auto moving_right_value = left_sibling_ptr->ValueAt(left_sibling_size - 1);
        left_sibling_ptr->Delete(left_sibling_size - 1);
        node_ptr->Insert(0, moving_up_key, moving_right_value);
        node_ptr->SetKeyAt(1, parent_key);
        parent_ptr->SetKeyAt(node_index, moving_up_key);
      }
    }
  }
}

/*****************************************************************************
 * UTILITIES AND DEBUG
 *****************************************************************************/

/*
 * This method is used for test only
 * Read data from file and insert one by one
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::InsertFromFile(const std::string &file_name, Transaction *txn) {
  int64_t key;
  std::ifstream input(file_name);
  while (input) {
    input >> key;

    KeyType index_key;
    index_key.SetFromInteger(key);
    RID rid(key);
    Insert(index_key, rid, txn);
  }
}
/*
 * This method is used for test only
 * Read data from file and remove one by one
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::RemoveFromFile(const std::string &file_name, Transaction *txn) {
  int64_t key;
  std::ifstream input(file_name);
  while (input) {
    input >> key;
    KeyType index_key;
    index_key.SetFromInteger(key);
    Remove(index_key, txn);
  }
}

INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::Print(BufferPoolManager *bpm) {
  auto root_page_id = GetRootPageId();
  auto guard = bpm->FetchPageBasic(root_page_id);
  PrintTree(guard.PageId(), guard.template As<BPlusTreePage>());
}

INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::PrintTree(page_id_t page_id, const BPlusTreePage *page) {
  if (page->IsLeafPage()) {
    auto *leaf = reinterpret_cast<const LeafPage *>(page);
    std::cout << "Leaf Page: " << page_id << "\tNext: " << leaf->GetNextPageId() << std::endl;

    // Print the contents of the leaf page.
    std::cout << "Contents: ";
    for (int i = 0; i < leaf->GetSize(); i++) {
      std::cout << leaf->KeyAt(i);
      if ((i + 1) < leaf->GetSize()) {
        std::cout << ", ";
      }
    }
    std::cout << std::endl;
    std::cout << std::endl;

  } else {
    auto *internal = reinterpret_cast<const InternalPage *>(page);
    std::cout << "Internal Page: " << page_id << std::endl;

    // Print the contents of the internal page.
    std::cout << "Contents: ";
    for (int i = 0; i < internal->GetSize(); i++) {
      std::cout << internal->KeyAt(i) << ": " << internal->ValueAt(i);
      if ((i + 1) < internal->GetSize()) {
        std::cout << ", ";
      }
    }
    std::cout << std::endl;
    std::cout << std::endl;
    for (int i = 0; i < internal->GetSize(); i++) {
      auto guard = bpm_->FetchPageBasic(internal->ValueAt(i));
      PrintTree(guard.PageId(), guard.template As<BPlusTreePage>());
    }
  }
}

/**
 * This method is used for debug only, You don't need to modify
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::Draw(BufferPoolManager *bpm, const std::string &outf) {
  if (IsEmpty()) {
    LOG_WARN("Drawing an empty tree");
    return;
  }

  std::ofstream out(outf);
  out << "digraph G {" << std::endl;
  auto root_page_id = GetRootPageId();
  auto guard = bpm->FetchPageBasic(root_page_id);
  ToGraph(guard.PageId(), guard.template As<BPlusTreePage>(), out);
  out << "}" << std::endl;
  out.close();
}

/**
 * This method is used for debug only, You don't need to modify
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::ToGraph(page_id_t page_id, const BPlusTreePage *page, std::ofstream &out) {
  std::string leaf_prefix("LEAF_");
  std::string internal_prefix("INT_");
  if (page->IsLeafPage()) {
    auto *leaf = reinterpret_cast<const LeafPage *>(page);
    // Print node name
    out << leaf_prefix << page_id;
    // Print node properties
    out << "[shape=plain color=green ";
    // Print data of the node
    out << "label=<<TABLE BORDER=\"0\" CELLBORDER=\"1\" CELLSPACING=\"0\" CELLPADDING=\"4\">\n";
    // Print data
    out << "<TR><TD COLSPAN=\"" << leaf->GetSize() << "\">P=" << page_id << "</TD></TR>\n";
    out << "<TR><TD COLSPAN=\"" << leaf->GetSize() << "\">"
        << "max_size=" << leaf->GetMaxSize() << ",min_size=" << leaf->GetMinSize() << ",size=" << leaf->GetSize()
        << "</TD></TR>\n";
    out << "<TR>";
    for (int i = 0; i < leaf->GetSize(); i++) {
      out << "<TD>" << leaf->KeyAt(i) << "</TD>\n";
    }
    out << "</TR>";
    // Print table end
    out << "</TABLE>>];\n";
    // Print Leaf node link if there is a next page
    if (leaf->GetNextPageId() != INVALID_PAGE_ID) {
      out << leaf_prefix << page_id << " -> " << leaf_prefix << leaf->GetNextPageId() << ";\n";
      out << "{rank=same " << leaf_prefix << page_id << " " << leaf_prefix << leaf->GetNextPageId() << "};\n";
    }
  } else {
    auto *inner = reinterpret_cast<const InternalPage *>(page);
    // Print node name
    out << internal_prefix << page_id;
    // Print node properties
    out << "[shape=plain color=pink ";  // why not?
    // Print data of the node
    out << "label=<<TABLE BORDER=\"0\" CELLBORDER=\"1\" CELLSPACING=\"0\" CELLPADDING=\"4\">\n";
    // Print data
    out << "<TR><TD COLSPAN=\"" << inner->GetSize() << "\">P=" << page_id << "</TD></TR>\n";
    out << "<TR><TD COLSPAN=\"" << inner->GetSize() << "\">"
        << "max_size=" << inner->GetMaxSize() << ",min_size=" << inner->GetMinSize() << ",size=" << inner->GetSize()
        << "</TD></TR>\n";
    out << "<TR>";
    for (int i = 0; i < inner->GetSize(); i++) {
      out << "<TD PORT=\"p" << inner->ValueAt(i) << "\">";
      if (i > 0) {
        out << inner->KeyAt(i);
      } else {
        out << " ";
      }
      out << "</TD>\n";
    }
    out << "</TR>";
    // Print table end
    out << "</TABLE>>];\n";
    // Print leaves
    for (int i = 0; i < inner->GetSize(); i++) {
      auto child_guard = bpm_->FetchPageBasic(inner->ValueAt(i));
      auto child_page = child_guard.template As<BPlusTreePage>();
      ToGraph(child_guard.PageId(), child_page, out);
      if (i > 0) {
        auto sibling_guard = bpm_->FetchPageBasic(inner->ValueAt(i - 1));
        auto sibling_page = sibling_guard.template As<BPlusTreePage>();
        if (!sibling_page->IsLeafPage() && !child_page->IsLeafPage()) {
          out << "{rank=same " << internal_prefix << sibling_guard.PageId() << " " << internal_prefix
              << child_guard.PageId() << "};\n";
        }
      }
      out << internal_prefix << page_id << ":p" << child_guard.PageId() << " -> ";
      if (child_page->IsLeafPage()) {
        out << leaf_prefix << child_guard.PageId() << ";\n";
      } else {
        out << internal_prefix << child_guard.PageId() << ";\n";
      }
    }
  }
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::DrawBPlusTree() -> std::string {
  if (IsEmpty()) {
    return "()";
  }

  PrintableBPlusTree p_root = ToPrintableBPlusTree(GetRootPageId());
  std::ostringstream out_buf;
  p_root.Print(out_buf);

  return out_buf.str();
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::ToPrintableBPlusTree(page_id_t root_id) -> PrintableBPlusTree {
  auto root_page_guard = bpm_->FetchPageBasic(root_id);
  auto root_page = root_page_guard.template As<BPlusTreePage>();
  PrintableBPlusTree proot;

  if (root_page->IsLeafPage()) {
    auto leaf_page = root_page_guard.template As<LeafPage>();
    proot.keys_ = leaf_page->ToString();
    proot.size_ = proot.keys_.size() + 4;  // 4 more spaces for indent

    return proot;
  }

  // draw internal page
  auto internal_page = root_page_guard.template As<InternalPage>();
  proot.keys_ = internal_page->ToString();
  proot.size_ = 0;
  for (int i = 0; i < internal_page->GetSize(); i++) {
    page_id_t child_id = internal_page->ValueAt(i);
    PrintableBPlusTree child_node = ToPrintableBPlusTree(child_id);
    proot.size_ += child_node.size_;
    proot.children_.push_back(child_node);
  }

  return proot;
}

template class BPlusTree<GenericKey<4>, RID, GenericComparator<4>>;

template class BPlusTree<GenericKey<8>, RID, GenericComparator<8>>;

template class BPlusTree<GenericKey<16>, RID, GenericComparator<16>>;

template class BPlusTree<GenericKey<32>, RID, GenericComparator<32>>;

template class BPlusTree<GenericKey<64>, RID, GenericComparator<64>>;

}  // namespace bustub
